package net.opentsdb.core;

import org.hbase.async.Bytes;

import java.util.Comparator;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author hujf
 * @Date 2022/2/28 15:42
 * @Description 压缩队列 不断压缩
 */
public class CompactionQueue extends ConcurrentSkipListMap <byte[] ,Boolean> {
    private final AtomicInteger size = new AtomicInteger();
    private final TSDB tsdb;
    private final short metric_width;

    private final int flush_interval;
    private final int flush_speed;
    private final int max_concurrent_flushes;
    private final int min_flush_threshold;

    public CompactionQueue(TSDB tsdb) {
        super(new Cmp(tsdb));
        this.tsdb = tsdb;
        metric_width = tsdb.metrics.width();
        flush_interval = tsdb.getConfig().getInt("tsd.storage.compaction.flush_interval");
        min_flush_threshold = tsdb.getConfig().getInt("tsd.storage.compaction.min_flush_threshold");
        max_concurrent_flushes = tsdb.getConfig().getInt("tsd.storage.compaction.max_concurrent_flushes");
        flush_speed = tsdb.getConfig().getInt("tsd.storage.compaction.flush_speed");

        if(tsdb.getConfig().enable_compaction()){
            startCompactionThread();
        }
    }

    private void startCompactionThread() {
      //  final Thrd thrd = new Thrd();
      //  thrd.setDaemon(true);
       // thrd.start();
    }


    final class Thrd extends  Thread{
        public Thrd(){
            super("CompactionThread");
        }

        @Override
        public void run() {
            while(true){

                final int size = size();
                if(size>min_flush_threshold){
                    final int maxflushes = Math.max(min_flush_threshold,size*flush_interval*flush_speed/Const.MAX_TIMESPAN);
                    final long now = System.currentTimeMillis();

                    flush(now/1000-Const.MAX_TIMESPAN-1,maxflushes);

                }



            }
        }
    }

    private void flush(long l, int maxflushes) {
    }


    public void add(byte[] row) {
        try {
            if (super.put(row, Boolean.TRUE == null)) {
                size.incrementAndGet();
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    private static final class  Cmp implements Comparator<byte[]>{

        private final short timestamp_pos;
        public Cmp(TSDB tsdb) {
            timestamp_pos = (short) (Const.SALT_WIDTH()+tsdb.metrics.width());
        }

        @Override
        public int compare(byte[] a, byte[] b) {
            final int c = Bytes.memcmp(a,b,timestamp_pos,Const.TIMESTAMP_BYTES);
            return c!=0?c:Bytes.memcmp(a,b);
        }
    }
}
